package com.navi.job;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
//import net.sf.json.JSONObject;
import com.mysql.jdbc.log.LogUtils;
import com.navi.entity.ThreadUtil;
import com.navi.exception.HmsException;
import com.navi.job.service.MainCallable;
import com.navi.job.service.MainService;
import com.navi.service.impl.RabbitService;
import com.navi.util.*;
import com.rabbitmq.client.*;
import com.rabbitmq.http.client.domain.QueueInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.io.IOException;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;

import static com.navi.constant.GenericDef.*;
import static java.util.stream.Collectors.toList;

/**
 * Created by lin.xin on 2017/10/27.
 */
@Slf4j
@Service
public class HmsLoaderService {
    @Value("${gp.datasource.drive}")
    private String driver;
    @Value("${gp.datasource.url}")
    private String url;
    @Value("${gp.datasource.username}")
    private String username;
    @Value("${gp.datasource.password}")
    private String password;
    @Value("${spring.rabbitmq.exchange}")
    private String exchange;
    @Value("${spring.rabbitmq.error.queue.routekey}")
    private String errorQueueRouteKey;
    @Value("${spring.rabbitmq.error.queue.routekey2}")
    private String errorQueueRouteKey2;
    @Value("${hms.mode}")
    private String mode;

    @Autowired
    private ConnectionFactory connectionFactory;
    @Autowired
    private JedisPool jedisPool;
    @Autowired
    private HmsLoaderInfoService hmsLoaderInfoService;


    @Bean
    public TaskScheduler taskScheduler() {
        return new ConcurrentTaskScheduler(); //single threaded by default
    }

    private boolean testJedisConnection() {
        return !jedisPool.isClosed();
    }


    public boolean subMainProc(String queueName, int limit, String limit_time, String deleteInsert) {
        final long beginTime = System.currentTimeMillis();

        //从RabbitMQ队列中读取消息存入List<String>列表
        //按照表名称分组==> Map<String,List<JsonObject>> tableJsonListMap

        List<String> messageList = new ArrayList<>();

        //测试Redis连接是否可用
        final boolean canConnected = testJedisConnection();
        if (!canConnected) {
            log.error("Redis 连接不可用，退出程式，等待下一次运行");
            return canConnected;
        }
        //测试DB是否可用 TODO
        log.info("MQ连接中... Q-name {}", queueName);
        Map<GetResponse, String> responseMap = new HashMap<>();


        try (Connection connection = connectionFactory.newConnection()) {
            log.info("MQ连接成功...  Q-name {}", queueName);

            long lbeginTime = System.currentTimeMillis();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(exchange, ExchangeTypes.TOPIC, true);


            try {
                /**
                 * 在取出了队列中所有消息后，在将消息转成JsonObject时出现了异常 是很麻烦的事情
                 * 抛出异常后，程式将进入catch分支流程
                 * (因为将数据写入GreenPlum的逻辑 已经有try catch，所以并不会抛出异常)
                 */


                String queueNames = hmsLoaderInfoService.getQueuesNameList(queueName);
                log.info("开始从 queueName=[{}] 各取 {} 笔数据 ...", StringUtils.isNotEmpty(queueNames) ? queueNames : queueName, limit);
                //从MQ 里取数据
                if (StringUtils.isNotEmpty(queueNames)) {//多只Q
                    for (String queue_name : Arrays.asList(queueNames.split(","))) {
                        //TODO: 增加站点开关的逻辑。
                        List<String> messageListTemp = new ArrayList<>();
                        Map<GetResponse, String> responseMapTemp = new HashMap<>();
                        messageListTemp = (limit == -1)
                                ? hmsLoaderInfoService.getMessageFromQueue(channel, queue_name, responseMapTemp)
                                : hmsLoaderInfoService.getMessageFromQueue(channel, queue_name, limit, responseMapTemp);
                        messageList.addAll(messageListTemp);
                        responseMap.putAll(responseMapTemp);
                    }
                } else {
                    messageList = (limit == -1)
                            ? hmsLoaderInfoService.getMessageFromQueue(channel, queueName, responseMap)
                            : hmsLoaderInfoService.getMessageFromQueue(channel, queueName, limit, responseMap);
                }
                if (messageList.isEmpty()) {
                    log.info("queueName=[{}] 没有数据.", StringUtils.isNotEmpty(queueNames) ? queueNames : queueName);
                    return true;
                }

                log.info("queueName=[{}] 连接队列并获取消息, 消息数量：{},cost:{}ms", StringUtils.isNotEmpty(queueNames) ? queueNames : queueName, messageList.size(), System.currentTimeMillis() - lbeginTime);
                lbeginTime = System.currentTimeMillis();

                Map<String, List<JSONObject>>  tableJsonObjectListMap= null;
//                try (Jedis jedis = jedisPool.getResource()) {
                try{
                    tableJsonObjectListMap = hmsLoaderInfoService.getMessageFromQueueThenGroupByTableName2Map(messageList,responseMap);
                    if (null == tableJsonObjectListMap || tableJsonObjectListMap.size() == 0) {
                        log.info("queueName=[{}] 中没有能处理的数据:{}", StringUtils.isNotEmpty(queueNames) ? queueNames : queueName);
                    }
                } catch (Exception e) {
                    log.error("queueName=[{}] 将 数据 根据 ope_tbl_name 分组 转换成数组失败，失败原因：\n{}"
                            , StringUtils.isNotEmpty(queueNames) ? queueNames : queueName, LogUtil.stackTraceToString(e));
                }


                log.info("queueName=[" + (StringUtils.isNotEmpty(queueNames) ? queueNames : queueName) + "] 将 数据 根据 ope_tbl_name 分组 ，消息组数量：" + messageList.size() + "，cost:" + (System.currentTimeMillis() - lbeginTime) + "ms.");
//            lbeginTime=System.currentTimeMillis();


                String jobBatchId = UUID.randomUUID().toString();
                List<MainCallable> mainCallableListPdsGlass = new ArrayList<>();
                List<MainCallable> mainCallableListPdsSummary = new ArrayList<>();
                List<MainCallable> mainCallableListMdsGlass = new ArrayList<>();
                List<MainCallable> mainCallableListMdsSummary = new ArrayList<>();
                List<MainCallable> mainCallableListMdsResult = new ArrayList<>();
                List<MainCallable> mainCallableListCommon = new ArrayList<>();
                HashMap<String, String> summaryCheck = new HashMap<>();//common 是正常，pds 是制成，mds是量测
                //根据配置项执行对应的逻辑， insert or merge
                String finalQueueName = queueName;


                HashMap<String, List<JSONObject>> summaryMap = new HashMap<>();
                tableJsonObjectListMap.forEach((tableName, jsonObjectList) -> {
//                if(StringUtils.isEmpty(tableName)){
//                    log.info("错误数据{}",jsonObjectList);
//                }
//                else
                    if ("pds_glass_summary".equals(tableName)) {
                        summaryCheck.put(PDS, PDS);
                        hmsLoaderInfoService.getSummaryMap(PDS, jsonObjectList, summaryMap);
                    } else if ("mds_glass_summary".equals(tableName)) {
                        summaryCheck.put(MDS, MDS);
                        hmsLoaderInfoService.getSummaryMap(MDS, jsonObjectList, summaryMap);
                    } else {
                        summaryCheck.put(COMMON, COMMON);
                        final MainCallable mainCallable = new MainCallable(finalQueueName, tableName, jsonObjectList, channel, responseMap, limit_time, deleteInsert, jobBatchId);
                        mainCallableListCommon.add(mainCallable);
                    }
                });

                /*** 提交线程，并得到线程的回调 **/
                if (!summaryCheck.isEmpty() && null != summaryCheck.get(COMMON)) {
                    //ThreadCnt
                    Integer futureTaskListCommon = ThreadUtil.getThreadResult(COMMON, mainCallableListCommon);
                    while (futureTaskListCommon==RETURN_REWORK_ALLCASE){
                        log.info("[{}]线程任务 出现 Redis异常，重跑当前线程。",COMMON);
                        futureTaskListCommon = ThreadUtil.getThreadResult(COMMON, mainCallableListCommon);
                    }
                    if(futureTaskListCommon==RETURN_FAISE) throw new Exception("出现未处理异常，请呼叫工程师处理。");

                    log.info("Main Message: {}, Table对应的线程数量: {}, Que数量: {}, 总共花费时间:{} ms", "线程任务执行完毕...",
                            futureTaskListCommon + "/" + tableJsonObjectListMap.size(), messageList.size(), (System.currentTimeMillis() - beginTime));
                } else if (!summaryCheck.isEmpty() && null != summaryCheck.get(PDS)) {
                    //pdsSummary
                    if (!summaryMap.get("pdsSummary").isEmpty()) {
                        summaryMap.get("pdsSummary").stream().collect(Collectors.groupingBy(jsonObj -> jsonObj.getString("ope_tbl_name"))).forEach(
                                (tableName1, jsonObjectList1) -> {
                                    final MainCallable mainCallablePdsSummary = new MainCallable(finalQueueName, tableName1, jsonObjectList1, channel, responseMap, limit_time, deleteInsert, jobBatchId);
                                    mainCallableListPdsSummary.add(mainCallablePdsSummary);
                                }
                        );


                        Integer futureTaskListPdsSummary = ThreadUtil.getThreadResult("PDS SUMMARY", mainCallableListPdsSummary);

                        while (futureTaskListPdsSummary==RETURN_REWORK_ALLCASE){
                            log.info("[{}]线程任务 出现 Redis异常，重跑当前线程。","PDS SUMMARY");
                            futureTaskListPdsSummary = ThreadUtil.getThreadResult("PDS SUMMARY", mainCallableListPdsSummary);
                        }
                        if(futureTaskListPdsSummary==RETURN_FAISE) throw new Exception("出现未处理异常，请呼叫工程师处理。");

                        log.info("Main Message SUMMARY: {}, Table对应的线程数量: {}, Que数量: {}, 总共花费时间:{} ms", "线程任务执行完毕...",
                                futureTaskListPdsSummary + "/" + tableJsonObjectListMap.size(), messageList.size(), (System.currentTimeMillis() - beginTime));

                    }
                    //pdsGlass
                    summaryMap.get("pdsGlass").stream().collect(Collectors.groupingBy(jsonObj -> jsonObj.getString("ope_tbl_name"))).forEach(
                            (tableName1, jsonObjectList1) -> {
                                final MainCallable mainCallablePdsGlass = new MainCallable(finalQueueName, tableName1, jsonObjectList1, channel, responseMap, limit_time, deleteInsert, jobBatchId);
                                mainCallableListPdsGlass.add(mainCallablePdsGlass);
                            }
                    );
                    Integer futureTaskListPdsGlass = ThreadUtil.getThreadResult("PDS GLASS", mainCallableListPdsGlass);
                    while (futureTaskListPdsGlass==RETURN_REWORK_ALLCASE){
                        log.info("[{}]线程任务 出现 Redis异常，重跑当前线程。","PDS GLASS");
                        futureTaskListPdsGlass = ThreadUtil.getThreadResult("PDS GLASS", mainCallableListPdsGlass);
                    }
                    if(futureTaskListPdsGlass==RETURN_FAISE) throw new Exception("出现未处理异常，请呼叫工程师处理。");


                    log.info("Main Message GLASS: {}, Table对应的线程数量: {}, Que数量: {}, 总共花费时间:{} ms", "线程任务执行完毕...",
                            futureTaskListPdsGlass + "/" + tableJsonObjectListMap.size(), messageList.size(), (System.currentTimeMillis() - beginTime));

                } else if (!summaryCheck.isEmpty() && null != summaryCheck.get(MDS)) {
                    //mdsResult
                    if (!summaryMap.get("mdsResult").isEmpty()) {
                        summaryMap.get("mdsResult").stream().collect(Collectors.groupingBy(jsonObj -> jsonObj.getString("ope_tbl_name"))).forEach(
                                (tableName1, jsonObjectList1) -> {
                                    final MainCallable mainCallableMdsResult = new MainCallable(finalQueueName, tableName1, jsonObjectList1, channel, responseMap, limit_time, deleteInsert, jobBatchId);
                                    mainCallableListMdsResult.add(mainCallableMdsResult);
                                }
                        );
                        Integer futureTaskListMdsResult = ThreadUtil.getThreadResult("MDS RESULT", mainCallableListMdsResult);
                        while (futureTaskListMdsResult==RETURN_REWORK_ALLCASE){
                            log.info("[{}]线程任务 出现 Redis异常，重跑当前线程。","MDS RESULT");
                            futureTaskListMdsResult = ThreadUtil.getThreadResult("MDS RESULT", mainCallableListMdsResult);
                        }
                        if(futureTaskListMdsResult==RETURN_FAISE) throw new Exception("出现未处理异常，请呼叫工程师处理。");

                        log.info("Main Message Result: {}, Table对应的线程数量: {}, Que数量: {}, 总共花费时间:{} ms", "线程任务执行完毕...",
                                futureTaskListMdsResult + "/" + tableJsonObjectListMap.size(), messageList.size(), (System.currentTimeMillis() - beginTime));

                    }
                    //mdsSummary
                    if (!summaryMap.get("mdsSummary").isEmpty()) {
                        summaryMap.get("mdsSummary").stream().collect(Collectors.groupingBy(jsonObj -> jsonObj.getString("ope_tbl_name"))).forEach(
                                (tableName1, jsonObjectList1) -> {
                                    final MainCallable mainCallableMdsSummary = new MainCallable(finalQueueName, tableName1, jsonObjectList1, channel, responseMap, limit_time, deleteInsert, jobBatchId);
                                    mainCallableListMdsSummary.add(mainCallableMdsSummary);
                                }
                        );
                        Integer futureTaskListMdsSummary = ThreadUtil.getThreadResult("MDS SUMMARY", mainCallableListMdsSummary);
                        while (futureTaskListMdsSummary==RETURN_REWORK_ALLCASE){
                            log.info("[{}]线程任务 出现 Redis异常，重跑当前线程。","MDS SUMMARY");
                            futureTaskListMdsSummary = ThreadUtil.getThreadResult("MDS SUMMARY", mainCallableListMdsSummary);
                        }
                        if(futureTaskListMdsSummary==RETURN_FAISE) throw new Exception("出现未处理异常，请呼叫工程师处理。");

                        log.info("Main Message SUMMARY: {}, Table对应的线程数量: {}, Que数量: {}, 总共花费时间:{} ms", "线程任务执行完毕...",
                                futureTaskListMdsSummary + "/" + tableJsonObjectListMap.size(), messageList.size(), (System.currentTimeMillis() - beginTime));

                    }
                    //mdsGlass
                    summaryMap.get("mdsGlass").stream().collect(Collectors.groupingBy(jsonObj -> jsonObj.getString("ope_tbl_name"))).forEach(
                            (tableName1, jsonObjectList1) -> {
                                final MainCallable mainCallableMdsGlass = new MainCallable(finalQueueName, tableName1, jsonObjectList1, channel, responseMap, limit_time, deleteInsert, jobBatchId);
                                mainCallableListMdsGlass.add(mainCallableMdsGlass);
                            }
                    );
                    Integer futureTaskListMdsGlass = ThreadUtil.getThreadResult("MDS GLASS", mainCallableListMdsGlass);
                    while (futureTaskListMdsGlass==RETURN_REWORK_ALLCASE){
                        log.info("[{}]线程任务 出现 Redis异常，重跑当前线程。","MDS GLASS");
                        futureTaskListMdsGlass = ThreadUtil.getThreadResult("MDS GLASS", mainCallableListMdsGlass);
                    }
                    if(futureTaskListMdsGlass==RETURN_FAISE) throw new Exception("出现未处理异常，请呼叫工程师处理。");

                    log.info("Main Message GLASS: {}, Table对应的线程数量: {}, Que数量: {}, 总共花费时间:{} ms", "线程任务执行完毕...",
                            futureTaskListMdsGlass + "/" + tableJsonObjectListMap.size(), messageList.size(), (System.currentTimeMillis() - beginTime));
                }

                return true;
                //结束:打印耗时，写入行数等性能参数
            } catch (Exception ex) {
                log.error(LogUtil.stackTraceToString(ex));
                return false;
//                while(1==1) {
//                    log.info("等待工程师处理异常中...(Log 2min 打印一次,处理完毕请重启程式)");
//                    Thread.sleep(120000);
//                }
//                try {
//                    // todo 这里送的是取出来的全部消息，没有经过转JSON或者转换时出现异常， 这部分没有转换成对象，所以无法记录进入Error的明细
//                    sendStringToErrorQueue2(errorQueueRouteKey, messageList);
//                    // todo ack
//                    //确定消费
//                    new MainService().sureAck(channel, responseMap);
//                    return false;
//                } catch (Exception e) {
//                    log.error(e.toString());
//                    LogUtil.pringStackTrace(e);
//                }
            }
        } catch (Exception e) {
            log.error(LogUtil.stackTraceToString(e));
            return false;
        }
//        return false;

    }


    private void sendStringToErrorQueue2(String routeKey, List<String> list) throws IOException, TimeoutException {
        if (CollectionUtils.isEmpty(list)) {
            return;
        }
        try (Connection connection = connectionFactory.newConnection()) {
            final Channel channel = connection.createChannel();
            list.forEach(message -> sendMessage(routeKey, channel, message));
        } catch (Exception e) {
            LogUtil.pringStackTrace(e);
            throw new HmsException(e);
        }
    }

    private void sendMessage(String routeKey, Channel channel, String message) {
        try {
            channel.basicPublish(exchange, routeKey,
                    new AMQP.BasicProperties.Builder()
                            .contentType("text/plain")
                            .deliveryMode(2)//设置消息持久性:2永不过期
                            .priority(1).build(), message.getBytes());
        } catch (IOException e) {
            LogUtil.pringStackTrace(e);
            throw new HmsException(e);
        }
    }


}
